//! 连接管理器
//! 
//! 管理网络连接的生命周期，维护连接ID映射，支持pipeline命令处理

// use std::collections::HashMap; // 暂时未使用
use std::sync::{Arc, Mutex, atomic::{AtomicU64, Ordering}};
use std::thread;
use std::time::{Duration, Instant};
use fxhash::FxHashMap;

use crate::protocol::NetworkStream;
// use crate::storage::Command; // 暂时未使用
use crate::event::{ConnectionId, RequestId};

/// 连接状态
#[derive(Debug, Clone, PartialEq)]
pub enum ConnectionState {
    /// 连接中
    Connected,
    /// 正在关闭
    Closing,
    /// 已关闭
    Closed,
}

/// 连接信息
#[derive(Debug, Clone)]
pub struct ConnectionInfo {
    /// 连接ID
    pub id: ConnectionId,
    /// 连接状态
    pub state: ConnectionState,
    /// 远程地址
    pub remote_addr: Option<String>,
    /// 创建时间
    pub created_at: Instant,
    /// 最后活跃时间
    pub last_active: Instant,
    /// 处理的命令数量
    pub command_count: u64,
    /// 待处理的请求数量
    pub pending_requests: u64,
}

/// 待处理请求信息
#[derive(Debug)]
pub struct PendingRequest {
    /// 请求ID
    pub request_id: RequestId,
    /// 连接ID
    pub connection_id: ConnectionId,
    /// 请求创建时间
    pub created_at: Instant,
    /// 命令名称（用于统计）
    pub command_name: String,
}

/// 连接管理器
/// 
/// 负责：
/// 1. 管理所有网络连接的生命周期
/// 2. 为每个连接分配唯一ID
/// 3. 维护连接到NetworkStream的映射
/// 4. 支持pipeline命令处理
/// 5. 连接状态监控和清理
pub struct ConnectionManager {
    /// 连接ID生成器
    next_connection_id: AtomicU64,
    /// 请求ID生成器（全局）
    next_request_id: AtomicU64,
    /// 活跃连接映射: ConnectionId -> (NetworkStream, ConnectionInfo)
    connections: Arc<Mutex<FxHashMap<ConnectionId, (Box<dyn NetworkStream>, ConnectionInfo)>>>,
    /// 待处理请求映射: RequestId -> PendingRequest
    pending_requests: Arc<Mutex<FxHashMap<RequestId, PendingRequest>>>,
    /// 连接统计信息
    total_connections: AtomicU64,
    /// 总处理请求数
    total_requests: AtomicU64,
}

impl ConnectionManager {
    /// 创建新的连接管理器
    pub fn new() -> Self {
        Self {
            next_connection_id: AtomicU64::new(1),
            next_request_id: AtomicU64::new(1),
            connections: Arc::new(Mutex::new(FxHashMap::default())),
            pending_requests: Arc::new(Mutex::new(FxHashMap::default())),
            total_connections: AtomicU64::new(0),
            total_requests: AtomicU64::new(0),
        }
    }

    /// 注册新连接
    pub fn register_connection(&self, stream: Box<dyn NetworkStream>) -> ConnectionId {
        let connection_id = self.next_connection_id.fetch_add(1, Ordering::Relaxed);
        let remote_addr = stream.remote_addr();
        
        let connection_info = ConnectionInfo {
            id: connection_id,
            state: ConnectionState::Connected,
            remote_addr: remote_addr.clone(),
            created_at: Instant::now(),
            last_active: Instant::now(),
            command_count: 0,
            pending_requests: 0,
        };
        
        {
            let mut connections = self.connections.lock().unwrap();
            connections.insert(connection_id, (stream, connection_info));
        }
        
        self.total_connections.fetch_add(1, Ordering::Relaxed);
        
        println!("📡 注册新连接 ID: {} 来自: {:?}", 
            connection_id, 
            remote_addr.unwrap_or_else(|| "unknown".to_string())
        );
        
        connection_id
    }

    /// 注销连接
    pub fn unregister_connection(&self, connection_id: ConnectionId) -> bool {
        let mut connections = self.connections.lock().unwrap();
        
        if let Some((mut stream, mut info)) = connections.remove(&connection_id) {
            info.state = ConnectionState::Closed;
            
            // 尝试优雅关闭连接
            if let Err(e) = stream.shutdown() {
                eprintln!("⚠️ 关闭连接 {connection_id} 时发生错误: {e}");
            }
            
            println!("📡 注销连接 ID: {} 处理了 {} 个命令", connection_id, info.command_count);
            true
        } else {
            false
        }
    }

    /// 生成新的请求ID
    pub fn generate_request_id(&self) -> RequestId {
        self.next_request_id.fetch_add(1, Ordering::Relaxed)
    }

    /// 记录待处理请求
    pub fn add_pending_request(&self, connection_id: ConnectionId, command_name: String) -> RequestId {
        let request_id = self.generate_request_id();
        
        let pending_request = PendingRequest {
            request_id,
            connection_id,
            created_at: Instant::now(),
            command_name,
        };
        
        {
            let mut pending = self.pending_requests.lock().unwrap();
            pending.insert(request_id, pending_request);
        }
        
        // 更新连接信息
        {
            let mut connections = self.connections.lock().unwrap();
            if let Some((_, info)) = connections.get_mut(&connection_id) {
                info.pending_requests += 1;
                info.last_active = Instant::now();
            }
        }
        
        self.total_requests.fetch_add(1, Ordering::Relaxed);
        request_id
    }

    /// 完成请求处理
    pub fn complete_request(&self, request_id: RequestId) -> Option<PendingRequest> {
        let mut pending = self.pending_requests.lock().unwrap();
        let request = pending.remove(&request_id);
        
        if let Some(ref req) = request {
            // 更新连接统计
            let mut connections = self.connections.lock().unwrap();
            if let Some((_, info)) = connections.get_mut(&req.connection_id) {
                info.command_count += 1;
                info.pending_requests = info.pending_requests.saturating_sub(1);
                info.last_active = Instant::now();
            }
        }
        
        request
    }

    /// 向连接写入响应数据
    pub fn write_response(&self, connection_id: ConnectionId, response_data: &[u8]) -> Result<usize, String> {
        let mut connections = self.connections.lock().unwrap();
        
        match connections.get_mut(&connection_id) {
            Some((stream, info)) => {
                match stream.write(response_data) {
                    Ok(bytes_written) => {
                        stream.flush().map_err(|e| format!("刷新缓冲区失败: {e}"))?;
                        info.last_active = Instant::now();
                        Ok(bytes_written)
                    }
                    Err(e) => {
                        // 写入失败，标记连接为关闭状态
                        info.state = ConnectionState::Closing;
                        Err(format!("写入连接 {connection_id} 失败: {e}"))
                    }
                }
            }
            None => Err(format!("连接 {connection_id} 不存在")),
        }
    }

    /// 从连接读取数据
    pub fn read_from_connection(&self, connection_id: ConnectionId, buffer: &mut [u8]) -> Result<usize, String> {
        let mut connections = self.connections.lock().unwrap();
        
        match connections.get_mut(&connection_id) {
            Some((stream, info)) => {
                match stream.read(buffer) {
                    Ok(bytes_read) => {
                        if bytes_read > 0 {
                            info.last_active = Instant::now();
                        }
                        Ok(bytes_read)
                    }
                    Err(e) => {
                        // 读取失败，可能连接已断开
                        info.state = ConnectionState::Closing;
                        Err(format!("从连接 {connection_id} 读取失败: {e}"))
                    }
                }
            }
            None => Err(format!("连接 {connection_id} 不存在")),
        }
    }

    /// 检查连接是否仍然活跃
    pub fn is_connection_active(&self, connection_id: ConnectionId) -> bool {
        let connections = self.connections.lock().unwrap();
        
        match connections.get(&connection_id) {
            Some((stream, info)) => {
                info.state == ConnectionState::Connected && stream.is_connected()
            }
            None => false,
        }
    }

    /// 获取连接信息
    pub fn get_connection_info(&self, connection_id: ConnectionId) -> Option<ConnectionInfo> {
        let connections = self.connections.lock().unwrap();
        connections.get(&connection_id).map(|(_, info)| (*info).clone())
    }

    /// 获取所有活跃连接ID列表
    pub fn get_active_connections(&self) -> Vec<ConnectionId> {
        let connections = self.connections.lock().unwrap();
        connections
            .iter()
            .filter(|(_, (stream, info))| {
                info.state == ConnectionState::Connected && stream.is_connected()
            })
            .map(|(id, _)| *id)
            .collect()
    }

    /// 清理非活跃连接
    pub fn cleanup_inactive_connections(&self, timeout: Duration) -> usize {
        let mut connections = self.connections.lock().unwrap();
        let now = Instant::now();
        let mut to_remove = Vec::new();
        
        for (id, (stream, info)) in connections.iter() {
            // 检查连接状态和超时
            if info.state != ConnectionState::Connected 
                || !stream.is_connected()
                || now.duration_since(info.last_active) > timeout {
                to_remove.push(*id);
            }
        }
        
        let removed_count = to_remove.len();
        for id in to_remove {
            if let Some((mut stream, _)) = connections.remove(&id) {
                let _ = stream.shutdown(); // 忽略关闭错误
                println!("🧹 清理非活跃连接: {id}");
            }
        }
        
        removed_count
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> ConnectionManagerStats {
        let connections = self.connections.lock().unwrap();
        let pending = self.pending_requests.lock().unwrap();
        
        let active_connections = connections
            .iter()
            .filter(|(_, (stream, info))| {
                info.state == ConnectionState::Connected && stream.is_connected()
            })
            .count();
        
        ConnectionManagerStats {
            total_connections: self.total_connections.load(Ordering::Relaxed),
            active_connections,
            pending_requests: pending.len() as u64,
            total_requests_processed: self.total_requests.load(Ordering::Relaxed),
        }
    }

    /// 打印统计信息
    pub fn print_stats(&self) {
        let stats = self.get_stats();
        println!("📊 连接管理器统计:");
        println!("  总连接数: {}", stats.total_connections);
        println!("  活跃连接数: {}", stats.active_connections);
        println!("  待处理请求: {}", stats.pending_requests);
        println!("  总处理请求数: {}", stats.total_requests_processed);
    }

    /// 启动后台清理任务
    pub fn start_cleanup_task(&self, cleanup_interval: Duration, connection_timeout: Duration) -> thread::JoinHandle<()> {
        let connections = Arc::clone(&self.connections);
        let pending = Arc::clone(&self.pending_requests);
        
        thread::Builder::new()
            .name("connection-cleanup".to_string())
            .spawn(move || {
                loop {
                    thread::sleep(cleanup_interval);
                    
                    // 清理超时连接
                    let now = Instant::now();
                    let mut connections_guard = connections.lock().unwrap();
                    let mut pending_guard = pending.lock().unwrap();
                    
                    let mut to_remove_connections = Vec::new();
                    let mut to_remove_requests = Vec::new();
                    
                    // 检查连接超时
                    for (id, (stream, info)) in connections_guard.iter() {
                        if info.state != ConnectionState::Connected 
                            || !stream.is_connected()
                            || now.duration_since(info.last_active) > connection_timeout {
                            to_remove_connections.push(*id);
                        }
                    }
                    
                    // 检查请求超时（防止内存泄漏）
                    for (id, req) in pending_guard.iter() {
                        if now.duration_since(req.created_at) > Duration::from_secs(300) { // 5分钟超时
                            to_remove_requests.push(*id);
                        }
                    }
                    
                    // 执行清理
                    for id in to_remove_connections {
                        if let Some((mut stream, _)) = connections_guard.remove(&id) {
                            let _ = stream.shutdown();
                        }
                    }
                    
                    for id in to_remove_requests {
                        pending_guard.remove(&id);
                    }
                }
            })
            .expect("创建连接清理线程失败")
    }
}

/// 连接管理器统计信息
#[derive(Debug, Clone)]
pub struct ConnectionManagerStats {
    /// 总连接数（历史）
    pub total_connections: u64,
    /// 当前活跃连接数
    pub active_connections: usize,
    /// 待处理请求数
    pub pending_requests: u64,
    /// 总处理请求数
    pub total_requests_processed: u64,
}

impl Default for ConnectionManager {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::NetworkStream;
    use std::io;

    // 模拟网络流用于测试
    #[derive(Debug)]
    struct MockNetworkStream {
        data: Vec<u8>,
        pos: usize,
        connected: bool,
    }

    impl MockNetworkStream {
        fn new(data: Vec<u8>) -> Self {
            Self {
                data,
                pos: 0,
                connected: true,
            }
        }
    }

    impl NetworkStream for MockNetworkStream {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            if self.pos >= self.data.len() {
                return Ok(0);
            }
            
            let to_copy = std::cmp::min(buf.len(), self.data.len() - self.pos);
            buf[..to_copy].copy_from_slice(&self.data[self.pos..self.pos + to_copy]);
            self.pos += to_copy;
            Ok(to_copy)
        }
        
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            Ok(buf.len())
        }
        
        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
        
        fn is_connected(&self) -> bool {
            self.connected
        }
        
        fn remote_addr(&self) -> Option<String> {
            Some("127.0.0.1:12345".to_string())
        }
        
        fn shutdown(&mut self) -> io::Result<()> {
            self.connected = false;
            Ok(())
        }
    }

    #[test]
    fn test_connection_registration() {
        let manager = ConnectionManager::new();
        let stream = Box::new(MockNetworkStream::new(vec![]));
        
        let conn_id = manager.register_connection(stream);
        assert_eq!(conn_id, 1);
        
        let info = manager.get_connection_info(conn_id).unwrap();
        assert_eq!(info.id, conn_id);
        assert_eq!(info.state, ConnectionState::Connected);
    }

    #[test]
    fn test_request_lifecycle() {
        let manager = ConnectionManager::new();
        let stream = Box::new(MockNetworkStream::new(vec![]));
        let conn_id = manager.register_connection(stream);
        
        // 添加请求
        let req_id = manager.add_pending_request(conn_id, "GET".to_string());
        assert_eq!(req_id, 1);
        
        // 完成请求
        let completed = manager.complete_request(req_id).unwrap();
        assert_eq!(completed.request_id, req_id);
        assert_eq!(completed.connection_id, conn_id);
        assert_eq!(completed.command_name, "GET");
    }

    #[test]
    fn test_connection_cleanup() {
        let manager = ConnectionManager::new();
        let stream = Box::new(MockNetworkStream::new(vec![]));
        let conn_id = manager.register_connection(stream);
        
        // 注销连接
        assert!(manager.unregister_connection(conn_id));
        assert!(!manager.is_connection_active(conn_id));
        
        // 尝试再次注销应该失败
        assert!(!manager.unregister_connection(conn_id));
    }

    #[test]
    fn test_statistics() {
        let manager = ConnectionManager::new();
        let stream = Box::new(MockNetworkStream::new(vec![]));
        let conn_id = manager.register_connection(stream);
        
        let req_id = manager.add_pending_request(conn_id, "SET".to_string());
        
        let stats = manager.get_stats();
        assert_eq!(stats.total_connections, 1);
        assert_eq!(stats.active_connections, 1);
        assert_eq!(stats.pending_requests, 1);
        
        manager.complete_request(req_id);
        let stats = manager.get_stats();
        assert_eq!(stats.pending_requests, 0);
        assert_eq!(stats.total_requests_processed, 1);
    }
}